package org.myfram.flink.flinkonjar.common.connector.oracle.statement;

import org.apache.flink.connector.jdbc.JdbcStatementBuilder;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.util.Arrays;
import java.util.List;

/**
 * 通用泛型 JDBC StatementBuilder，用于将 POJO 写入数据库
 */
public class AbstractGenericStatementBuilder<T> implements JdbcStatementBuilder<T> {

    private final List<String> fieldOrder;

    public AbstractGenericStatementBuilder(String... fieldOrder) {
        this.fieldOrder = Arrays.asList(fieldOrder);
    }

    @Override
    public void accept(PreparedStatement ps, T t) {
        Class<?> clazz = t.getClass();

        for (int i = 0; i < fieldOrder.size(); i++) {
            String fieldName = fieldOrder.get(i);
            Field field = null;
            try {
                field = clazz.getDeclaredField(fieldName);

                field.setAccessible(true);
                Object value = field.get(t);
                ps.setObject(i + 1, value); // JDBC 下标从 1 开始
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    }
}
